package com.message.queue.rocket.translation;

import com.message.queue.constant.RocketConstant;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.io.UnsupportedEncodingException;
import java.util.concurrent.*;

/**
 * Created by 张波 on 2019/10/19.
 */
public class TransProduct {

    public static final String TOPIC = "trans_topic3";


    public static final String TAGS = "trans_tag";

    public static void main(String[] args) throws MQClientException, UnsupportedEncodingException, InterruptedException {
        TransactionMQProducer product = new TransactionMQProducer("trans_product_group");
        ExecutorService executorService = new ThreadPoolExecutor(5, 100, 30, TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                /*创建回调方法线程*/
                Thread thread = new Thread(r);
                thread.setName("Thread check" );
                return thread;
            }
        });

        product.setNamesrvAddr(RocketConstant.NAME_SERVER);
        product.setExecutorService(executorService);

        TransactionListener transactionListener = new TransactionListenerImpl();
        product.setTransactionListener(transactionListener);

        product.start();
        for (int i = 0; i < 10; i++) {
            Message message = new Message(TOPIC, TAGS, "message" + i, ("Hello Message" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            product.sendMessageInTransaction(message, "" + i);//回调参数可在业务方法和回调方法中获取到，可为null
        }
        Thread.sleep(Long.MAX_VALUE);
        product.shutdown();
    }
}
